package org.example.using;


import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FilterFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.SparkSession;
import org.example.bean.AdsPhmFault;
import org.example.bean.TPhmGzzdYjgzbjRank;
import org.example.utils.MysqlUtils_2;

import java.io.IOException;

import static org.apache.spark.sql.functions.col;


/***
 * 创建一个类,用来测试 spark 2.3.0版本的 DataFrame的用法, DataSet API的使用.
 */
public class Spark03_SparkDataSetUnionUsing {
    public static void main(String[] args) throws IOException {
        // 初始化SparkSession
        // 引入log4j,消除控制台的警告
        Logger.getLogger("org").setLevel(Level.WARN);
        Logger.getLogger("akka").setLevel(Level.WARN);
        SparkConf conf = new SparkConf().setAppName("PantographModelFunction") // 设置应用程序名称
                .setMaster("local[*]") // 设置运行模式，这里是本地模式
                .set("spark.executor.memory", "6g") // 设置executor内存大小
                .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") // 设置序列化器
                .set("spark.driver.allowMultipleContexts", "true") // 允许多个SparkContext同时存在
                .set("spark.hadoop.validateOutputSpecs", "false") // 禁止输出路径已经存在时抛出异常
                .set("hive.mapred.supports.subdirectories", "true") // 开启Hive支持
                .set("mapreduce.input.fileinputformat.input.dir.recursive", "true") // 开启递归读取文件夹中的文件
                .set("spark.debug.maxToStringFields", "100").set("spark.sql.codegen.wholeStage", "false")
                .set("spark.sql.caseSensitive", "false"); // Spark会自动处理大小写不匹配的情况
        JavaSparkContext sc = new JavaSparkContext(conf);
        SparkSession spark = SparkSession.builder().config(conf).getOrCreate();
        /*
            这个案例中,主要介绍了两个或者多个DataSet,进行 union 、except、intersect
         */

        Dataset<TPhmGzzdYjgzbjRank> tPhmGzzdYjgzbjRankDS = MysqlUtils_2.readMySQLTableDF(spark, "t_phm_gzzd_yjgzbj_rank", TPhmGzzdYjgzbjRank.class);
        Dataset<TPhmGzzdYjgzbjRank> accDS = tPhmGzzdYjgzbjRankDS.filter(
                (FilterFunction<TPhmGzzdYjgzbjRank>) tPhmGzzdYjgzbjRank -> tPhmGzzdYjgzbjRank.getGzmc_group_id().equals("ACC")
        );
        Dataset<TPhmGzzdYjgzbjRank> l1DS = tPhmGzzdYjgzbjRankDS.filter(
                (FilterFunction<TPhmGzzdYjgzbjRank>) tPhmGzzdYjgzbjRank -> tPhmGzzdYjgzbjRank.getGzmc_group_id().equals("L1")
        );

        // intersect 获取两个集合的交集
        tPhmGzzdYjgzbjRankDS.intersect(accDS).show(false);
        // 获取两个数据集的并集,也就是获取长表.
        accDS.union(l1DS).show(false);
        // 获取两个数据集的差集,也就是 tPhmGzzdYjgzbjRankDS 中 不包含 accDS的部分.
        tPhmGzzdYjgzbjRankDS.except(accDS).show(false);
    }
}
